package com.sunzm.flink.datastream.scala.operator

import org.apache.commons.lang3.StringUtils
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 *
 * ScalaOperatorDemo
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-06-22 20:47
 */
object ScalaOperatorDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {

    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    //方法一
   /* val wordcountDS: DataStream[(String, Int)] = dataStream.flatMap(line => {
      if(StringUtils.isNotBlank(line)){
        //切分后的单词数组
        line.split(",")
      }else{
        //空数组
        Array.empty[String]
      }
    }).map((_, 1))
      .keyBy(_._1)
      .sum(1)*/

    //方法二
    /**
     * 总结： 使用带 colletor（收集器）的flatMap和不带收集器的flatMap方法的区别
     *
     * 普通的 flatMap 方法，必须要有返回值，即使是一个空的，也要构造一个空集合返回
     * 而带 收集器的 flatMap 方法，输出数据是靠收集器来控制的，所以可以不输出数据
     */
    val wordcountDS: DataStream[(String, Int)] = dataStream.flatMap((line, colletor: Collector[(String, Int)]) => {

      if(StringUtils.isNotBlank(line)){
        val fields = line.split(",")
        fields.foreach(word => {
          colletor.collect((word, 1))
        })
      }

    }).keyBy(_._1)
      .sum(1)

    wordcountDS.print()

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
